package com.opages.netty.demo.netty.rpc.consumer;

import com.opages.netty.demo.netty.rpc.common.RpcInvoker;
import com.opages.netty.demo.netty.rpc.common.RpcProtocol;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.concurrent.Callable;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @author daizhong.liu
 * @create 2021-12-15 17:35
 **/
public class ConsumerHandler extends ChannelInboundHandlerAdapter implements Callable {
    private ChannelHandlerContext ctx;
    private RpcInvoker invoker;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;
    }

    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //读取结果集
        if(msg != null && msg instanceof RpcProtocol) {
            RpcProtocol protocol = (RpcProtocol) msg;
            RpcInvoker invokerTmp = protocol.toInvoker();
            this.invoker.setResult(invokerTmp.getResult());
        }
        notify();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }

    @Override
    public synchronized Object call() throws Exception {
        ctx.writeAndFlush(invoker.toProtocol());
        System.err.println();
        wait();
        return invoker.getResult();
    }

    public RpcInvoker getInvoker() {
        return invoker;
    }

    public void setInvoker(RpcInvoker invoker) {
        this.invoker = invoker;
    }
}
